package com.ruoyi.uitity;


import com.ruoyi.framework.config.RabbitConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.concurrent.*;

@Component
public class WxMessageProducer {
    private static final Logger log = LoggerFactory.getLogger(WxMessageProducer.class);
    @Autowired
    private RabbitTemplate rabbitTemplate;
    @Autowired
    private RabbitAdmin rabbitAdmin;

    private int corePoolSize = 10;

    private int maxPoolSize= 20;

    private int queueCapacity = 50;

    private ExecutorService executor;

    @PostConstruct
    public void init() {
        executor = new ThreadPoolExecutor(
                corePoolSize,
                maxPoolSize,
                60L, TimeUnit.SECONDS,
                new LinkedBlockingQueue<>(queueCapacity),
                new ThreadPoolExecutor.CallerRunsPolicy()
        );
    }

    public Future<String> sendMessage(String msg, Long userId) {

        rabbitAdmin.purgeQueue(RabbitConfig.QUEUE_NAME, false);
        // 创建一个与 userId 关联的 CorrelationData 对象
        CorrelationData correlationId = new CorrelationData(userId.toString());

        // 使用线程池异步发送消息
        Future<String> future = executor.submit(() -> {
            String response = (String) rabbitTemplate.convertSendAndReceive(RabbitConfig.EXCHANGE_NAME, RabbitConfig.ROUTING_KEY, msg, correlationId);
            // 等待响应
            return response;
        });
        return future;
    }

    @PreDestroy
    public void destroy() {
       log.info("优雅关闭问答线程池");
        executor.shutdown();
        try {
            if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
                executor.shutdownNow();
            }
        } catch (InterruptedException e) {
            executor.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }
}
